package spark;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.streaming.Durations;
import org.apache.spark.streaming.api.java.JavaInputDStream;
import org.apache.spark.streaming.api.java.JavaPairDStream;
import org.apache.spark.streaming.api.java.JavaStreamingContext;
import org.apache.spark.streaming.kafka010.ConsumerStrategies;
import org.apache.spark.streaming.kafka010.KafkaUtils;
import org.apache.spark.streaming.kafka010.LocationStrategies;

import java.util.Arrays;
import java.util.Collection;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Map;

import scala.Tuple2;

/**
 * @author zhaoxuan
 * @date 2021-09-03 15:49
 **/
public class SparkMain {
    public static final long STREAMING_BATCH_INTERVAL=5L;
    public static void main(String[] args) throws InterruptedException {
        SparkConf conf = new SparkConf().setMaster("local").setAppName("wordCountTest");
        JavaSparkContext sc = new JavaSparkContext(conf);
        sc.setLogLevel("WARN");
        JavaStreamingContext jsc = new JavaStreamingContext(sc, Durations.seconds(STREAMING_BATCH_INTERVAL));
        String topic = "my-replicated-topic";
        String brokers = "hadoop01:9092";
        Collection<String> topics = new HashSet<>(Arrays.asList(topic.split(",")));
        //kafka相关参数，必要！缺了会报错
        Map<String, Object> kafkaParams = new HashMap<>();
        kafkaParams.put("bootstrap.servers", brokers);
        kafkaParams.put("group.id", "group1");
        //kafkaParams.put("key.serializer", StringSerializer.class);
        kafkaParams.put("key.deserializer", StringDeserializer.class);
        kafkaParams.put("value.deserializer", StringDeserializer.class);
        JavaInputDStream<ConsumerRecord<Object, Object>> lines = KafkaUtils.createDirectStream(jsc, LocationStrategies.PreferConsistent(), ConsumerStrategies.Subscribe(topics, kafkaParams));
        JavaPairDStream<String, Integer> count = lines
                .flatMap(x -> Arrays.asList(x.value().toString().split(" ")).iterator())
                .mapToPair(x -> new Tuple2<>(x, 1))
                .window(Durations.seconds(STREAMING_BATCH_INTERVAL*3),Durations.seconds(STREAMING_BATCH_INTERVAL*3))
                .reduceByKey(Integer::sum);

        count.print();
        //lines.foreachRDD(rdd-> rdd.foreach(x-> System.out.println(x.value())));
        jsc.start();
        jsc.awaitTermination();
        jsc.close();
/*        SparkSession sparkSession = SparkSession.builder().appName("sparksqlTest").config("spark.some.config.option", "some-value").getOrCreate();
        Dataset<Row> dataset = sparkSession.read().json("cloud-security/src/main/resources/data.json");
        dataset.createOrReplaceTempView("test");
        sparkSession.sql("select aaa from test").show();*/

        //dataset.show();
    }
}
